#!/usr/bin/python
# -*- coding: utf-8 -*-

import os
import sys
import time
import SocketServer
import threading
import thread
from optparse import OptionParser
import logging
import logging.handlers

import htlib
from htlib import Atomic

class Operation(threading._Timer):
    def __init__(self, *args, **kwargs):
        threading._Timer.__init__(self, *args, **kwargs)
        self.setDaemon(True)

    def run(self):
        while True:
            self.finished.clear()
            self.finished.wait(self.interval)
            if not self.finished.isSet():
                self.function(*self.args, **self.kwargs)
            else:
                return
            self.finished.set()

class Schedule(object):
    ops = []

    def add(self, operation, interval, args=[], kwargs={}):
        self.op = Operation(interval, operation, args, kwargs)
        self.ops.append(self.op)

    def start(self):
        for op in self.ops:
            thread.start_new_thread(op.run, ())

    def stop(self):
        for op in self.ops:
            op.cancel()
        self._event.set()

class ClusterGroup:
    def __init__(self, group_id, total_clients):
        self.round = Atomic(0)
        self.count = Atomic(0)
        self.sense = 1
        self.total_clients = total_clients
        self._read_clients = Atomic(0)
        self.timestamp = Atomic(0.0)
        self.group_id = group_id
        self.not_wait = False

        self.timeout = 300
        #self.timeout = 3600
        """ @param timeout > concurrency * filesize / write ratio """

        htlib.log_out('Lock server group timeout is set to %d!!!\n' % self.timeout)
        mylogger.info('Lock server group timeout is set to %d!!!' % self.timeout)

    def __repr__(self):
        return 'group_id: %s, round %d, count %d, sense %d, total_clients %d, timestamp %f' % \
                (self.group_id, self.round.get(), self.cound.get(), self.sense, \
                self.total_clients, self.timestamp.get())


class SimpleServer(SocketServer.ThreadingMixIn, SocketServer.TCPServer):
    daemon_threads      = True
    allow_reuse_address = True
    request_queue_size  = 256

    def __init__(self, server_address, RequestHandlerClass):
        SocketServer.TCPServer.__init__(self, server_address, RequestHandlerClass)

        self.lock      = threading.Lock()
        self.groups    = {}
        self.checktime = 1.0

        self._watchdog = Schedule()
        self._watchdog.add(self.check_group, self.checktime)
        self._watchdog.start()

    def check_group(self):
        self.lock.acquire()
        for k, v in self.groups.items():
            if v.count.get() == 0:
                continue

            now = time.time()
            if now - v.timestamp.get() > v.timeout:
                # if less than a half clients 
                if v.count.get() <= v.total_clients/2:
                    self.groups[k].timeout = int(v.timeout * 1.5)
                    htlib.log_out('There are %d clients arrived, total clients is %d...\n' % (v.count.get(), v.total_clients))
                    htlib.log_out('Increment group timeout to %d.\n' % int(self.groups[k].timeout))
                    mylogger.info('There are %d clients arrived, total clients is %d...\n' % (v.count.get(), v.total_clients))
                    mylogger.info('Increment group timeout to %d.\n' % int(self.groups[k].timeout))
                    continue

                self.groups[k].not_wait = True
                self.groups[k].timestamp.set(time.time())
        self.lock.release()

class RequestHandler(SocketServer.BaseRequestHandler):
    """
    requests:
        - barrier
        - cancel
    response code:
		- 200 OK
		- 300 Warning
		- 400 Client Error
		- 500 Server Error
    """
    def __init__(self, request, client_address, server):
        SocketServer.BaseRequestHandler.__init__(self, request, client_address, server)
        self.client_address = client_address

    def handle(self):
        htlib.log_out('Get connect from %s %d\n' % (self.client_address[0], self.client_address[1]))
        mylogger.info('Get connect from %s %d' % (self.client_address[0], self.client_address[1]))

        try:
            # If client disconnected before send any data
            data = self.request.recv(1024)
            if not data:
                self.finish()
                return
            
            data = data.strip()

            # If Client asked server to delete its group
            if data.startswith('cancel#'):
                gid = data.split('#')[1].strip()

                self.server.lock.acquire()

                try:
                    if self.server.groups.has_key(gid):
                        del self.server.groups[gid]
                except KeyError:
                    htlib.log_err('Client %s del group id [%s] error\n' % (self.client_address[0], gid))
                    self.request.send('500#Cancel %s failed\n' % gid)
                    self.server.lock.release()
                    self.finish()
                    return
                else:
                    htlib.log_out('Client %s del group id [%s] success\n' % (self.client_address[0], gid))
                    self.request.send('200#Cancel %s successfully\n' % gid)
                    self.server.lock.release()
                    self.finish()
                    return
            elif data.startswith('barrier#'):
                pass
            else:
                return

            # data format 'barrier#group_id:round:total_clients:local_sense'
            params = data[len('barrier#'):].strip()
            group_id, next_round, total_clients, local_sense,  = [i.strip() for i in params.split(':') if i is not '']

            # verify round
            try:
                if int(next_round) != self.server.groups[group_id].round.get() + 1:
                    resp = '400#round mismatched: request round %d, current round %d!!!' % (int(next_round), self.server.groups[group_id].round.get())
                    htlib.log_err('Send %s to %s %d\n' % (resp, self.client_address[0], self.client_address[1]))
                    mylogger.info('Send %s to %s %d' % (resp, self.client_address[0], self.client_address[1]))
                    self.request.send(resp)
                    self.finish()
                    return
            except ValueError:
                return
            except KeyError:
                pass

            try:
                local_sense   = int(local_sense)
                total_clients = int(total_clients)

                self.server.lock.acquire()
                if not self.server.groups.has_key(group_id):
                    htlib.log_out('Can not found group [%s], add it \n' % group_id)
                    self.server.groups[group_id] = ClusterGroup(group_id, total_clients)
                
                # update timestamp
                self.server.groups[group_id].timestamp.set(time.time())

                # if the first client
                if self.server.groups[group_id].count.get() == 0:
                    self.server.groups[group_id].not_wait = False

                self.server.lock.release()
            except ValueError:
                htlib.log_err("Parse from %s error'\n" % str(self.client_address)) 
                exit(1)

            # Main arithmetic start here
            if self.server.groups[group_id].count.increment_and_get() == self.server.groups[group_id].total_clients:
                self.server.groups[group_id].count.set(0)
                self.server.groups[group_id].sense = local_sense
                round = self.server.groups[group_id].round.increment_and_get()

                htlib.log_out('All client are done!!!\n')
                htlib.log_out('Current round is [ %d ]\n' % round)
                htlib.log_out('#################### #_# ###################\n\n')

                mylogger.info('All client are done!!!')
                mylogger.info('Current round is [ %d ]' % round)
                mylogger.info('#################### #_# ###################')
            else:
                htlib.log_out('Current client is [ %d ] , total client is [ %d ]\n' % (self.server.groups[group_id].count.get(), self.server.groups[group_id].total_clients))
                mylogger.info('Current client is [ %d ] , total client is [ %d ]' % (self.server.groups[group_id].count.get(), self.server.groups[group_id].total_clients))
                while True:
                    try:
                        resp = '200#Ready to continue\n'

                        if self.server.groups[group_id].not_wait:
                            if self.server.groups[group_id].sense != local_sense:
                                self.server.groups[group_id]._read_clients.set(self.server.groups[group_id].count.get())
                                # update total_clients
                                self.server.groups[group_id].total_clients = self.server.groups[group_id].count.get()
                                self.server.groups[group_id].count.set(0)
                                self.server.groups[group_id].sense = local_sense
                                round = self.server.groups[group_id].round.increment_and_get()

                                htlib.log_out("Don't wait for other clients, let us go!!!\n")
                                htlib.log_out('Current round is [ %d ]\n' % round)
                                htlib.log_out('#################### #_# ###################\n\n')

                                mylogger.info("%s: Don't wait for other clients, let us go!!!" % time.ctime())
                                mylogger.info('%s: Current round is [ %d ]' % (time.ctime(), round))
                            # self.server.groups[group_id].not_wait = False
                            resp = '300#%d' % self.server.groups[group_id]._read_clients.get()
                            htlib.log_err('Warning: send %s to %s %d\n' % (resp, self.client_address[0], self.client_address[1]))
                            mylogger.info('Warning: send %s to %s %d\n' % (resp, self.client_address[0], self.client_address[1]))

                        if self.server.groups[group_id].sense != local_sense:
                            time.sleep(0.1)
                        else:
                            self.request.send(resp)
                            break
                    except KeyError:
                        resp = '301#no group %s\n' % group_id
                        self.request.send(resp)
                        break
        except:
            self.finish()
            raise
        self.finish()

    def finish(self):
        self.request.close()

class LockServer:
    def __init__(self, host='localhost', port=20008):
        self._addr     = (host, port)
        self._pid_file = '/tmp/lock_server/lock_server.pid'
        self._log_dir  = '/tmp/lock_server/'

    def _get_logpath(self):
        if not os.path.exists(self._log_dir):
            os.mkdir(self._log_dir)

        tm = time.strftime('%Y-%m-%d-%H:%M', time.gmtime())
        out_log = self._log_dir + tm + '.log'
        err_log = self._log_dir + tm + '.err'
        return (out_log, err_log)

    def start(self, daemonize=False):
        server = SimpleServer(self._addr, RequestHandler)

        if daemonize:
            self._clean_pidfile(self._pid_file)
            out_log, err_log = self._get_logpath()
            self._daemonize(self._pid_file, out_log, err_log)
            htlib.log_out('Start server with daemon mode\n')
            mylogger.info('Start server with daemon mode')
        else:
            htlib.log_out('Start server without daemon mode\n')
            mylogger.info('Start server without daemon mode')

        try:
            server.serve_forever()
        except KeyboardInterrupt:
            htlib.log_out('Get Key interrupt, exit...\n')
            sys.exit(1)
        except:
            raise
        self._clean_pidfile(self._pid_file)
        
    def _clean_pidfile(self, pid_file):
        try:
            os.unlink(pid_file)
        except OSError, e:
            # htlib.log_err("No such pid file %s\n" % pid_file)
            pass

    def _daemonize(self, pid_file, out_log, err_log):
        sys.stdout.flush()
        sys.stderr.flush()

        try:
            pid = os.fork()
            if pid > 0:
                sys.exit(0)
        except OSError, e:
            sys.stderr.write("fork daemon #1 failed: (%d) %s\n" % (e.errno, e.strerror))
            sys.exit(1)

        os.chdir('/')
        os.umask(0)
        os.setsid()

        try:
            pid = os.fork()
            if pid > 0:
                sys.exit(0)
        except OSError, e:
            sys.stderr.write("fork daemon #2 failed: (%d) %s\n" % (e.errno, e.strerror))
            sys.exit(1)

        pid = str(os.getpid())
        try:
            os.open(self._pid_file, os.O_WRONLY | os.O_CREAT | os.O_EXCL)
        except OSError, e:
            sys.stderr.write("Daemon already run\n") % sys.exit(1)

        open(pid_file, 'w').write(pid)

        si = open('/dev/null', 'r')
        so = open(out_log, 'a+', 1024)
        se = open(err_log, 'a+')

        os.dup2(si.fileno(), sys.stdin.fileno())
        os.dup2(so.fileno(), sys.stdout.fileno())
        os.dup2(se.fileno(), sys.stderr.fileno())
            
if __name__ == '__main__':
    parser = OptionParser()
    parser.add_option("-H", "--host", type="string", dest="host", help="bind lock service to this host ip", default="0.0.0.0")
    parser.add_option("-P", "--port", type="int", dest="port", help="bind lock service to this port", default="20008")
    parser.add_option("-d", "--daemon", dest='daemon', action='store_true', default=False)
    (options, args) = parser.parse_args()

    LOG_FILENAME = 'ls_%d.log' % int(time.time())
    mylogger = logging.getLogger('MyLogger')
    mylogger.setLevel(logging.DEBUG)
    handler = logging.handlers.RotatingFileHandler(LOG_FILENAME, maxBytes=100*1024*1024, backupCount=5)
    mylogger.addHandler(handler)

    my_lock = LockServer(options.host, options.port)
    my_lock.start(options.daemon)
